package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import io.netty.channel.ChannelHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.checkerframework.checker.units.qual.A;
import org.joda.time.DateTime;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private CacheService cacheService;



    @Autowired
    private Redisson redisson;

    /**
     * 添加定时任务
     *
     * @param task
     */
    @Override
    @Transactional
    public void saveTask(Task task) {

        //1、将任务添加到taskinfo表中
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskinfo);

        task.setTaskId(taskinfo.getTaskId());

        //2、将任务添加到taskinfo_logs表中(version=1,status=0)
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfoLogs);
        taskinfoLogs.setVersion(1);  //版本号，用作乐观锁
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        taskinfoLogsMapper.insert(taskinfoLogs);

        //3、判断任务是否到期，如果到期将任务放到list中
        if(System.currentTimeMillis()>=task.getExecuteTime()){
            //topic_1001_1
            String listKey=ScheduleConstants.TOPIC+task.getTaskType()+"_"+task.getPriority();
            cacheService.lLeftPush(listKey, JSON.toJSONString(task));
        }

        //4、如果即将到期(大于当前系统时间，小于当前系统时间+5分分钟)，放到zset中
        if(task.getExecuteTime()>System.currentTimeMillis()&&task.getExecuteTime()<= System.currentTimeMillis()+1000*60*5){
            //future_1001_1
            String setKey=ScheduleConstants.FUTURE+task.getTaskType()+"_"+task.getPriority();
            cacheService.zAdd(setKey, JSON.toJSONString(task),task.getExecuteTime());
        }
    }


    /**
     * 从list集合拉取已经到期的文章
     *
     * @param taskType
     * @param priority
     * @return
     */
    @Override
    @Transactional
    public Task poll(Integer taskType, Integer priority) {

        //构建list集合的key  topic_1001_1
        String listKey = ScheduleConstants.TOPIC+taskType+"_"+priority;

        //从key中移除并返回第一个记录
        String s = cacheService.lRightPop(listKey);

        if(StringUtils.isNotBlank(s)){
            Task task = JSON.parseObject(s, Task.class);

            //将任务从taskifno表中删除
            taskinfoMapper.deleteById(task.getTaskId());

            //将taskinfo表中的任务的状态改为已消费
            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
            taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
            taskinfoLogsMapper.updateById(taskinfoLogs);


            return task;
        }


        return null;
    }


    /**
     * 定时刷新zset中已到期的数据到list中
     */
    @Scheduled(cron = "*/5 * * * * ?")
    public void refreshZetToList(){

        RLock mylock = redisson.getLock("mylock");  //得到锁

        boolean b = mylock.tryLock();  //开始尝试获取锁，如果获取到则返回true,否则返回false

        if(b) {

            try {
                //setKeys=["future_1001_1","future_1002_2"]
                Set<String> setKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");  //future_*

                if (setKeys != null && setKeys.size() > 0) {

                    for (String setKey : setKeys) {

                        //setKey=future_1001_1

                        String[] split = setKey.split("future_"); //split=["","1001_1"]
                        //获取zset对应的list的键
                        String listKey = ScheduleConstants.TOPIC + split[1];  //listKey="topic_1001_1"

                        //从zset里面获取到已经过期的文章
                        //[{"taskid": ,"parameters":,"tasktype":,"executeTime"},{}]
                        Set<String> strings = cacheService.zRangeByScore(setKey, 0, System.currentTimeMillis());

                        //如果已经到期的文章
                        if (strings != null && strings.size() > 0) {
                            //要将文章移到list中
                            cacheService.lLeftPushAll(listKey, strings);
                            //要将文章众zset中移除
                            cacheService.zRemove(setKey, strings);
                        }

                    }
                }
            } catch (Exception ex) {
                ex.printStackTrace();

            } finally {

                mylock.unlock();   //释放锁
            }
        }


    }


    /**
     * 将数据库中的任务刷到list或Zset中
     */
    @Scheduled(cron = "0 */5 * * * ?")
    public void refreshDBToCache(){

        //清空缓存
        Set<String> setKey1 = cacheService.scan(ScheduleConstants.FUTURE);
        Set<String> listKey1 = cacheService.scan(ScheduleConstants.TOPIC);
        cacheService.delete(setKey1);
        cacheService.delete(listKey1);

        //1、查出taksinfo表中的所有数据
        List<Taskinfo> taskinfos = taskinfoMapper.selectList(null);

        if(taskinfos!=null && taskinfos.size()!=0){
            for (Taskinfo taskinfo : taskinfos) {
                if(taskinfo.getExecuteTime().getTime()<=System.currentTimeMillis()){
                    //topic_1001_1
                    String listKey=ScheduleConstants.TOPIC+taskinfo.getTaskType()+"_"+taskinfo.getPriority();

                    Task task = new Task();
                    BeanUtils.copyProperties(taskinfo,task);
                    task.setExecuteTime(taskinfo.getExecuteTime().getTime());

                    cacheService.lLeftPush(listKey, JSON.toJSONString(task));
                }

                if(taskinfo.getExecuteTime().getTime()>System.currentTimeMillis()&&taskinfo.getExecuteTime().getTime()<= System.currentTimeMillis()+1000*60*5){
                    //future_1001_1
                    String setKey=ScheduleConstants.FUTURE+taskinfo.getTaskType()+"_"+taskinfo.getPriority();

                    Task task = new Task();
                    BeanUtils.copyProperties(taskinfo,task);
                    task.setExecuteTime(taskinfo.getExecuteTime().getTime());

                    cacheService.zAdd(setKey, JSON.toJSONString(task),task.getExecuteTime());
                }
            }
        }
    }
}
